#include "adalog/appender/AsyncProxyAppender.h"
#include "adalog/LogAppender.h"
#include "adalog/buffer/BufferPoolManager.h"
#include "adalog/Logger.h"
#include "adalog/buffer/ReadBufferGuard.h"
#include <optional>

namespace adalog
{
    AsyncProxyAppender::AsyncProxyAppender(std::list<LogAppender::Ptr> appenders)
        : appenders_(appenders.begin(), appenders.end())
        , buffer_pool_(10)
        , thread_pool_(3)
        , consumer_(&AsyncProxyAppender::ConsumerThread, this)
    {
        for (int i = 0; i < buffer_pool_.Size(); ++i)
            free_buffer_queue_.Put(i);
    }

    AsyncProxyAppender::~AsyncProxyAppender() 
    {
        if (current_buffer_.IsValid())
            full_buffer_queue_.Put(current_buffer_.GetBufferId());
        stop_.store(true);
        consumer_.join();
    }

    void AsyncProxyAppender::Append(const char* data, size_t len)
    {
        if (!current_buffer_.IsValid())
        {
            size_t free_buffer_id = free_buffer_queue_.Get();
            auto opt = buffer_pool_.GetWriteBuffer(free_buffer_id);
            if (!opt.has_value())
                ADALOG_DEFAULT_ERROR("Can not get write buffer guard.");
            current_buffer_ = std::move(opt.value());
        }
        if (len > current_buffer_.WritableSize())
        {
            full_buffer_queue_.Put(current_buffer_.GetBufferId());
            size_t free_buffer_id = free_buffer_queue_.Get();
            auto opt = buffer_pool_.GetWriteBuffer(free_buffer_id);
            if (!opt.has_value())
                ADALOG_DEFAULT_ERROR("Can not get write buffer guard.");
            current_buffer_ = std::move(opt.value());
            current_buffer_.Reset();
            // 空闲的缓冲区若还是放不下就只能截断日志了
            if (len > current_buffer_.WritableSize())
                ADALOG_DEFAULT_ERROR("log record too long, it will be truncated.")
            len = std::min(current_buffer_.WritableSize(), len);
        }
        current_buffer_.WriteData(data, len);
    }

    void AsyncProxyAppender::ConsumerThread()
    {
        while(!stop_ || full_buffer_queue_.Size() != 0)
        {
            size_t full_buffer_id = full_buffer_queue_.Get();
            std::shared_ptr<std::atomic<int>> task_undo_count = std::make_shared<std::atomic<int>>(appenders_.size());
            for (auto& appender : appenders_)
            {
                thread_pool_.Enqueue([this, full_buffer_id, task_undo_count, appender](){
                    auto opt = buffer_pool_.GetReadBuffer(full_buffer_id);
                    if (!opt.has_value())
                        ADALOG_DEFAULT_ERROR("Can not get read buffer guard.");
                    auto read_guard = std::move(opt.value());
                    appender->Append(read_guard.GetDataBegin(), read_guard.ReadableSize());
                    if (task_undo_count->fetch_sub(1)  == 0)
                        free_buffer_queue_.Put(full_buffer_id);
                });
            }
        }
    }

} // namespace adalog